/*-------------------------------------------------------------------------
 *
 * pg_hot_data.c
 *	  for hot data precache
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"
#include "catalog/pg_hot_data.h"
#include "libpq-fe.h"
#include "lib/stringinfo.h"
#include "utils/timestamp.h"
#include "access/xlog.h"
#include "postmaster/postmaster.h"
#include <stdlib.h>

void PrecacheHotData()
{
    char instanceName[NAMEDATALEN]; //default:master
    char primaryHost[16];           //default:127.0.0.1
    char primaryUser[NAMEDATALEN];  //default:postgres
    char primaryPw[NAMEDATALEN];    //default:123456
    char primaryPort[8];            //default:PostPortNumber
    char localPort[8];              //default:master
    StringInfoData cmd, primaryConnStr, localConnStr;

    initStringInfo(&cmd);
    initStringInfo(&primaryConnStr);
    initStringInfo(&localConnStr);

    memset(instanceName, 0, NAMEDATALEN);
    memset(primaryHost, 0, 16);
    memset(primaryUser, 0, NAMEDATALEN);
    memset(primaryPw, 0, NAMEDATALEN);
    memset(primaryPort, 0, 8);
    memset(localPort, 0, 8);

    //parse
    if (strlen(PrimaryConnInfo) > 0)
    {
        char *temStr;
        char *temChr;
        int temStrLen;

        //instanceName
        temStr = strstr(PrimaryConnInfo, "application_name=");
        temStrLen = strlen("application_name=");
        
        if (temStr != NULL)
        {
            temChr = strchr(temStr, ' ');
            if (temChr != NULL)
            {
                memcpy(instanceName, temStr + temStrLen, temChr - temStr - temStrLen);
            }
            else
            {
                strcpy(instanceName, temStr + temStrLen);
            }
        }
        else
        {
            strcpy(instanceName, "master");
        }

        //primaryHost
        temStr = strstr(PrimaryConnInfo, "host=");
        temStrLen = strlen("host=");
        
        if (temStr != NULL)
        {
            temChr = strchr(temStr, ' ');
            if (temChr != NULL)
            {
                memcpy(primaryHost, temStr + temStrLen, temChr - temStr - temStrLen);
            }
            else
            {
                strcpy(primaryHost, temStr + temStrLen);
            }
        }
        else
        {
            strcpy(primaryHost, "127.0.0.1");
        }

        //primaryUser
        temStr = strstr(PrimaryConnInfo, "user=");
        temStrLen = strlen("user=");

        if (temStr != NULL)
        {
            temChr = strchr(temStr, ' ');
            if (temChr != NULL)
            {
                memcpy(primaryUser, temStr + temStrLen, temChr - temStr - temStrLen);
            }
            else
            {
                strcpy(primaryUser, temStr + temStrLen);
            }
        }
        else
        {
            strcpy(primaryUser, "postgres");
        }

        //primaryPw
        temStr = strstr(PrimaryConnInfo, "password=");
        temStrLen = strlen("password=");

        if (temStr != NULL)
        {
            temChr = strchr(temStr, ' ');
            if (temChr != NULL)
            {
                memcpy(primaryPw, temStr + temStrLen, temChr - temStr - temStrLen);
            }
            else
            {
                strcpy(primaryPw, temStr + temStrLen);
            }
        }
        else
        {
            strcpy(primaryPw, "123456");
        }

        //primaryPort
        temStr = strstr(PrimaryConnInfo, "port=");
        temStrLen = strlen("port=");

        if (temStr != NULL)
        {
            temChr = strchr(temStr, ' ');
            if (temChr != NULL)
            {
                memcpy(primaryPort, temStr + temStrLen, temChr - temStr - temStrLen);
            }
            else
            {
                strcpy(primaryPort, temStr + temStrLen);
            }
        }
        else
        {
            sprintf(primaryPort, "%d", PostPortNumber);
        }
    }
    else
    {
        strcpy(instanceName, "master");
        strcpy(primaryHost, "127.0.0.1");
        strcpy(primaryUser, "postgres");
        strcpy(primaryPw, "123456");
        sprintf(primaryPort, "%d", PostPortNumber);
    }

    //assemble primaryConnStr
    appendStringInfoString(&primaryConnStr, "host=");
    appendStringInfoString(&primaryConnStr, primaryHost);
    appendStringInfoString(&primaryConnStr, " user=");
    appendStringInfoString(&primaryConnStr, primaryUser);
    appendStringInfoString(&primaryConnStr, " password=");
    appendStringInfoString(&primaryConnStr, primaryPw);
    appendStringInfoString(&primaryConnStr, " port=");
    appendStringInfoString(&primaryConnStr, primaryPort);
    appendStringInfoString(&primaryConnStr, " dbname=postgres");

    //conn local
    sprintf(localPort, "%d", PostPortNumber);
    appendStringInfoString(&localConnStr, "host=127.0.0.1 port=");
    appendStringInfoString(&localConnStr, localPort);
    appendStringInfoString(&localConnStr, " user=postgres dbname=postgres");
    PGconn *localConn = PQconnectdb(localConnStr.data);
    if (PQstatus(localConn) != CONNECTION_OK)
    {
        PQfinish(localConn);
        //log
        return;
    }

    appendStringInfoString(&cmd, "SELECT datname, relname, crules FROM pg_hot_data WHERE crulessettime>cachetime AND clientname='");
    appendStringInfoString(&cmd, instanceName);
    appendStringInfoString(&cmd, "'");

    //Query the corresponding precache policy 
    PGresult *ruleRes = PQexec(localConn, cmd.data);
    if (PQresultStatus(ruleRes) != PGRES_TUPLES_OK)
    {
        PQclear(ruleRes);
        PQfinish(localConn);
        //log
        return;
    }
    int rows = PQntuples(ruleRes);
    for(int i=0; i<rows; i++)
    {
        char *datname;
        char *relname;
        char *crules;
        datname = PQgetvalue(ruleRes, i, 0);
        relname = PQgetvalue(ruleRes, i, 1);
        crules = PQgetvalue(ruleRes, i, 2);

        //precache hot data(table level)
        if (strcmp(crules, "t") == 0)
        {
            //precache
            resetStringInfo(&localConnStr);
            appendStringInfoString(&localConnStr, "host=127.0.0.1 port=");
            appendStringInfoString(&localConnStr, localPort);
            appendStringInfoString(&localConnStr, " user=postgres dbname=");
            appendStringInfoString(&localConnStr, datname);
            PGconn *precacheConn = PQconnectdb(localConnStr.data);
            if (PQstatus(precacheConn) != CONNECTION_OK)
            {
                PQfinish(precacheConn);
                //log
                continue;
            }
            resetStringInfo(&cmd);
            appendStringInfoString(&cmd, "precache select * from ");
            appendStringInfoString(&cmd, relname);

            PGresult *precacheRes = PQexec(precacheConn, cmd.data);
            if (PQresultStatus(precacheRes) != PGRES_TUPLES_OK)
            {
                PQclear(precacheRes);
                PQfinish(precacheConn);
                //log
                continue;
            }

            PQclear(precacheRes);
            PQfinish(precacheConn);

            //update primary pg_hot_data
            const char* currentTime = NULL;
            currentTime = timestamptz_to_str(GetCurrentTimestamp());
            resetStringInfo(&cmd);
            appendStringInfoString(&cmd, "UPDATE pg_hot_data SET cachetime='");
            appendStringInfoString(&cmd, currentTime);
            appendStringInfoString(&cmd, "' WHERE datname='");
            appendStringInfoString(&cmd, datname);
            appendStringInfoString(&cmd, "' AND relname='");
            appendStringInfoString(&cmd, relname);
            appendStringInfoString(&cmd, "' AND crules='");
            appendStringInfoString(&cmd, crules);
            appendStringInfoString(&cmd, "' AND clientname='");
            appendStringInfoString(&cmd, instanceName);
            appendStringInfoString(&cmd, "'");

            PGconn *primaryConn = PQconnectdb(primaryConnStr.data);
            if (PQstatus(primaryConn) != CONNECTION_OK)
            {
                PQfinish(primaryConn);
                //log
                continue;
            }
            PGresult *updateRes=PQexec(primaryConn, cmd.data);
            if (PQresultStatus(updateRes) != PGRES_TUPLES_OK)
            {
                PQclear(updateRes);
                PQfinish(primaryConn);
                //log
                continue;
            }
            PQclear(updateRes);
            PQfinish(primaryConn);
        }
    }
    PQclear(ruleRes);
    PQfinish(localConn);
}